package com.ytktt.study.week12.task6;

import com.ytktt.study.week12.task6.exception.KmqException;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Kmq缓冲区
 *
 * @author ytk
 */
public class KmqBuffer<T> {

    private final ReentrantLock lock;

    private final Object[] entries;

    private int putIndex;

    private int count;

    private final ConcurrentMap<String, Integer> takeIndexMap;

    private final Condition notEmpty;

    private final Condition notFull;

    public KmqBuffer(int capacity) {
        this.entries = new Object[capacity];
        this.lock = new ReentrantLock();
        this.notEmpty = lock.newCondition();
        this.notFull =  lock.newCondition();
        this.takeIndexMap = new ConcurrentHashMap<>(16);
    }

    public void registerConsumer(String name) {
        this.takeIndexMap.putIfAbsent(name, 0);
    }

    final T itemAt(int i) {
        return (T) this.entries[i];
    }

    private void enqueue(T x) {
        final Object[] items = this.entries;
        items[this.putIndex] = x;
        if (++this.putIndex == items.length) {
            this.putIndex = 0;
        }
        this.count++;
        this.notEmpty.signal();
    }

    public boolean offer(T message) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (this.count == entries.length) {
                return false;
            } else {
                this.enqueue(message);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

    public boolean offer(T message, long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (this.count == this.entries.length) {
                if (nanos <= 0) {
                    return false;
                }
                nanos = this.notFull.awaitNanos(nanos);
            }
            this.enqueue(message);
            return true;
        } finally {
            lock.unlock();
        }
    }

    public void put(T message) throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (this.count == this.entries.length) {
                this.notFull.await();
            }
            this.enqueue(message);
        } finally {
            lock.unlock();
        }
    }

    public T poll(String name) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return (this.count == 0) ? null : this.getMessage(name);
        } finally {
            lock.unlock();
        }
    }

    public T take(String name) throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (this.count == 0) {
                this.notEmpty.await();
            }
            return this.getMessage(name);
        } finally {
            lock.unlock();
        }
    }

    public T poll(String name, long timeout, TimeUnit unit) throws InterruptedException {
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (this.count == 0) {
                if (nanos <= 0) {
                    return null;
                }
                nanos = notEmpty.awaitNanos(nanos);
            }
            return this.getMessage(name);
        } finally {
            lock.unlock();
        }
    }

    public T peek(String name) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return getMessage(name);
        } finally {
            lock.unlock();
        }
    }

    public void seek(String name, int offset) {
        if (offset < 0) {
            throw new KmqException("offset must greater than 0");
        }
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (offset > this.count) {
                throw new KmqException("offset must smaller than count");
            }
            this.takeIndexMap.put(name, offset);
        } finally {
            lock.unlock();
        }
    }

    private T getMessage(String name) {
        Integer takeIndex = this.takeIndexMap.get(name);
        T message = this.itemAt(takeIndex);
        if (message != null) {
            this.takeIndexMap.put(name, ++takeIndex);
        }
        return message;
    }
}
